package com.atguigu.flink.chapter11;

import com.atguigu.flink.chapter05.Source.WaterSensor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/1/27 10:11
 */
public class Flink03_TableAPI_Connect_FileSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 通过 connect直接 将 外部系统 抽象成 动态表,作为数据源
        //  1. 调用 connect方法，传入一个外部系统的描述器，还有一些参数
        //  2. 调用 withFormat方法，指定数据的存储格式： 列分隔符、行分隔符，等等
        //  3. 调用 wichSchema方法，指定 表的结构信息：  列名、列的类型
        //  4. 调用 createTemporaryTable方法，创建一张临时表，并且指定表名

        Schema schema = new Schema()
                .field("a", DataTypes.STRING())
                .field("b", DataTypes.BIGINT())
                .field("c", DataTypes.INT());

        tableEnv
                .connect(new FileSystem().path("input/sensor-sql.txt"))
                .withFormat(new Csv().fieldDelimiter(',').lineDelimiter("\n"))
                .withSchema(schema)
                .createTemporaryTable("fileSourceTable");

        // 从 表名 获取 Table对象
        Table fileSourceTable = tableEnv.from("fileSourceTable");

        Table resultTable = fileSourceTable
                .groupBy($("a"))
                .select($("a"), $("c").sum().as("vcSum"));

        // 转成流、打印
        tableEnv.toRetractStream(resultTable, Row.class).print();


        env.execute();
    }
}
